package com.pk4pk.rxjava.learning;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.schedulers.Schedulers;
/**
 * 
 * @Author CaiWF
 * @Email  401885064@qq.com
 * @Date 2016年12月08日
 * @TODO
 */
public class RxJavaDemo {
	private final static String TAG = RxJavaDemo.class.getName();

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		RxJavaDemo rxJavaDemo = new RxJavaDemo();

		// rxJavaDemo.helloRxJava();
		// rxJavaDemo.testDefer(rxJavaDemo);
		// rxJavaDemo.testEmpty();
		// rxJavaDemo.testNever();
		// rxJavaDemo.testError();
		// rxJavaDemo.testForm();
		// rxJavaDemo.testInterval();
		// rxJavaDemo.testJust();
		// rxJavaDemo.testRange();
		// rxJavaDemo.testRepeat();
		// rxJavaDemo.testTimer();
//		rxJavaDemo.testBuffer();
		rxJavaDemo.testMap();
		
		
		// 因测试testInterval需要，不让这个Main进程结束掉
//		 rxJavaDemo.NotStop();
	}

	private void testMap(){
		Observable.just(1,2,3)
		.filter(new Func1<Integer, Boolean>() {

			@Override
			public Boolean call(Integer t) {
				// TODO Auto-generated method stub
				return t<3;
			}
				
		})
		.map(new Func1<Integer,Integer>() {

			@Override
			public Integer call(Integer t) {
				// TODO Auto-generated method stub
				return t;
			}
			
		})
		
		.map(new Func1<Integer,String>() {

			@Override
			public String call(Integer t) {
				// TODO Auto-generated method stub
				return "map :"+t;
			}
		})
		.subscribe(new Action1<String>() {

			@Override
			public void call(String t) {
				// TODO Auto-generated method stub
				System.out.println("t :"+t);
			}
		});

	}
	
	private void testBuffer() {
		// 一组缓存3个数据
		Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).buffer(3).subscribe(i -> Log.d(TAG, "1buffer-count:" + i));
		
		// 每隔三个数据缓存2个数据
		Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).buffer(2, 3)
				.subscribe(i -> Log.d(TAG, "1buffer-count&skip:" + i));

		//间隔发送测试
		Observable.interval(1, TimeUnit.SECONDS).buffer(3, TimeUnit.SECONDS)
				.subscribe(i -> Log.d(TAG, "2buffer-count:" + i));
		
		
		Observable.interval(1, TimeUnit.SECONDS).buffer(2, 3, TimeUnit.SECONDS)
				.subscribe(i -> Log.d(TAG, "2buffer-count&skip:" + i));
	}

	/**
	 * Timer
	 */
	private void testTimer() {
		SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
		String startTime = sdf.format(new Date());
		Log.d(TAG, "startTime:" + startTime);
		Observable.timer(2, TimeUnit.SECONDS).observeOn(Schedulers.newThread()).subscribe(new Subscriber<Long>() {
			@Override
			public void onNext(Long item) {
				// Timer创建的对象在2秒钟后发射了一个0
				Log.d(TAG, "Next: " + item);
				String endTime = sdf.format(new Date());
				Log.d(TAG, "endTime:" + endTime);
			}

			@Override
			public void onError(Throwable error) {
				System.err.println("Error: " + error.getMessage());
			}

			@Override
			public void onCompleted() {
				Log.d(TAG, "Sequence complete.");
			}
		});
	}

	/**
	 * Repeat
	 */
	private void testRepeat() {
		// 重复5次发送数据1
		Observable.just(1).repeat(5).subscribe(new Subscriber<Integer>() {
			@Override
			public void onNext(Integer item) {
				Log.d(TAG, "Next: " + item);
			}

			@Override
			public void onError(Throwable error) {
				System.err.println("Error: " + error.getMessage());
			}

			@Override
			public void onCompleted() {
				Log.d(TAG, "Sequence complete.");
			}
		});
	}

	/**
	 * Range
	 */
	private void testRange() {
		Observable.range(100, 6).subscribe(new Subscriber<Integer>() {
			@Override
			public void onNext(Integer item) {
				Log.d(TAG, "Next: " + item);
			}

			@Override
			public void onError(Throwable error) {
				System.err.println("Error: " + error.getMessage());
			}

			@Override
			public void onCompleted() {
				Log.d(TAG, "Sequence complete.");
			}
		});
	}

	/**
	 * Just
	 */
	private void testJust() {
		Observable.just(1, 2, 3).subscribe(new Subscriber<Integer>() {
			@Override
			public void onNext(Integer item) {
				Log.d(TAG, "Next: " + item);
			}

			@Override
			public void onError(Throwable error) {
				System.err.println("Error: " + error.getMessage());
			}

			@Override
			public void onCompleted() {
				Log.d(TAG, "Sequence complete.");
			}
		});
	}

	/**
	 * Interval
	 */
	private void testInterval() {
		// 以秒为单位，每隔1秒发射一个数据
		Observable.interval(1, TimeUnit.SECONDS)
				// interva operates by default on the computation Scheduler,so
				// observe on main Thread
				// 如果需要更新view，要在主线程中订阅
				// .observeOn(AndroidSchedulers.mainThread())
				// .observeOn(Schedulers.io())
				.subscribe(new Subscriber<Long>() {
					@Override
					public void onCompleted() {
						Log.d(TAG, "onCompleted");
					}

					@Override
					public void onError(Throwable e) {
						Log.d(TAG, "onError:" + e.getMessage());
					}

					@Override
					public void onNext(Long aLong) {
						Log.d(TAG, "interval:" + aLong);
						// textView.setText("Interval:"+aLong);
					}
				});
	}

	/**
	 * Form
	 */
	@SuppressWarnings({ "unchecked", "unused", "rawtypes" })
	private void testForm() {
		Integer[] items = { 0, 1, 2, 3, 4, 5 };
		Observable myObservable = Observable.from(items);
		myObservable.subscribe(new Action1<Integer>() {
			@Override
			public void call(Integer item) {
				Log.d(TAG, item + "");
			}
		}, new Action1<Throwable>() {
			@Override
			public void call(Throwable error) {
				Log.d(TAG, "Error encountered: " + error.getMessage());
			}
		}, new Action0() {
			@Override
			public void call() {
				Log.d(TAG, "Sequence complete");
			}
		});
	}

	/**
	 * Error
	 */
	private void testError() {
		// Error:创建一个不发射数据以一个错误终止的Observable（只会调用onError）
		Observable.error(new Throwable("just call onError")).subscribe(new Subscriber<Object>() {
			@Override
			public void onNext(Object item) {
				Log.d(TAG, "Error：Next: " + item);
			}

			@Override
			public void onError(Throwable error) {
				Log.d(TAG, "Error：Error: " + error.getMessage());
			}

			@Override
			public void onCompleted() {
				Log.d(TAG, "Error：Sequence complete.");
			}
		});
	}

	/**
	 * Never
	 */
	private void testNever() {
		// Never:创建一个不发射数据也不终止的Observable（不会调用订阅者的任何方法）
		Observable.never().doOnSubscribe(new Action0() {
			@Override
			public void call() {
				// TODO Auto-generated method stub
				Log.d(TAG, "doOnSubscribe...");
			}
		}).subscribe(new Subscriber<Object>() {
			@Override
			public void onNext(Object item) {
				Log.d(TAG, "Nerver：Next: " + item);
			}

			@Override
			public void onError(Throwable error) {
				Log.d(TAG, "Nerver：Error: " + error.getMessage());
			}

			@Override
			public void onCompleted() {
				Log.d(TAG, "Nerver：Sequence complete.");
			}
		});
	}

	/**
	 * Empty
	 */
	private void testEmpty() {
		// enpty默认实现call，只调用onCompleted：public void call(Subscriber<? super
		// Object> child) {child.onCompleted();}
		Observable.empty().subscribe(new Subscriber<Object>() {
			@Override
			public void onNext(Object item) {
				System.out.println("Enpty：Next: " + item);
			}

			@Override
			public void onError(Throwable error) {
				System.out.println("Enpty：Error: " + error.getMessage());
			}

			@Override
			public void onCompleted() {
				System.out.println("Enpty：Sequence complete.");
			}
		});
	}

	/**
	 * Defer
	 */
	private Observable<Date> deferObservable;
	private Observable<Date> justObservable;

	private void testDefer(RxJavaDemo rxJavaDemo) {
		try {
			rxJavaDemo.op_Defer();
			Thread.sleep(3 * 1000);
			rxJavaDemo.op_Defer();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	private void op_Defer() {

		if (deferObservable == null) {
			deferObservable = Observable.defer(() -> Observable.just(new Date()));
		}
		deferObservable.subscribe(date -> {
			SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
			System.out.println("defer1:" + sdf.format(date));
		});
		deferObservable.subscribe(date -> {
			SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
			System.out.println("defer2:" + sdf.format(date));
		});
		deferObservable.subscribe(date -> {
			SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
			System.out.println("defer3:" + sdf.format(date));
		});

		if (justObservable == null) {
			justObservable = Observable.just(new Date());
		}
		justObservable.subscribe(date -> {
			SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
			System.out.println("just:" + sdf.format(date));
		});
	}

	/**
	 * HelloRxJava
	 */
	private void helloRxJava() {
		Observable.just("hello").subscribe(new Action1<String>() {
			public void call(String str) {
				// TODO Auto-generated method stub
				System.out.println("onNext :" + str);
			}
		});
	}

	/**
	 * NotStop
	 */
	private void NotStop() {
		while (true) {
			try {
				Thread.sleep(100 * 1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

}
